package com.qxp.simplerpc.monitor.service.impl;

import com.qxp.simplerpc.entity.InvokerMessage;
import com.qxp.simplerpc.entity.MonitorMessage;
import com.qxp.simplerpc.monitor.service.MonitorService;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * @author qianxp
 */
@Component
public class MonitorServiceImpl implements MonitorService {
    // 缓存对象
    private static volatile Map<String,MonitorMessage> cacheMap = new ConcurrentHashMap<>();
    // 缓存对象 记录时间
    private static volatile Map<String,Long> cacheTimeMap = new ConcurrentHashMap<>();

    private static final Lock lock = new ReentrantLock();

    @Override
    public void accept(MonitorMessage target) {
        try {
            lock.lock();
            String token = target.getToken();
            if(!cacheMap.containsKey(token)){
                cacheMap.putIfAbsent(token,target);
            }else {
                MonitorMessage cacheMsg = cacheMap.get(token);
                setInvokeMessageIfNull(target, cacheMsg);
            }
        }finally {
            lock.unlock();
        }
    }

    @Override
    public List<MonitorMessage> getCachedInvokeMsg() {
         return cacheTimeMap.entrySet().stream()
                .sorted((stringLongEntry, t1) -> (int)(stringLongEntry.getValue() - t1.getValue()))
                .map(Map.Entry::getKey)
                .map(key-> cacheMap.get(key))
                .collect(Collectors.toList());

    }


    /**
     *
     * @param targetInvokerMsg the latest invokeMsg
     * @param shadeInvokerMsg the cachedList
     */
    private void setInvokeMessageIfNull(MonitorMessage targetInvokerMsg,MonitorMessage shadeInvokerMsg){
        // 获取属性
        Field[] field = MonitorMessage.class.getDeclaredFields();
        Arrays.asList(field).parallelStream().forEach(f->{
            f.setAccessible(true);
            try {
                Object targetField =f.get(targetInvokerMsg);
                Object shadeField = f.get(shadeInvokerMsg);
                if(targetField == null && shadeField != null){
                    f.set(targetInvokerMsg,shadeField);
                }
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        });
        cacheMap.put(targetInvokerMsg.getToken(),targetInvokerMsg);
        cacheTimeMap.put(targetInvokerMsg.getToken(),System.currentTimeMillis());
   }







    @PostConstruct
   private void clearCache(){
       ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
       executorService.schedule(()->{},30,TimeUnit.MINUTES);
       List<String> clearCacheTokenList = new CopyOnWriteArrayList<>();
       if(cacheTimeMap.isEmpty()){
           return;
       }
       cacheTimeMap.forEach((k,v)->{
            if(System.currentTimeMillis() - v > 1000*60*60*30){
                clearCacheTokenList.add(k);
                cacheMap.remove(k);
            }
       });
       clearCacheTokenList.parallelStream().forEach(cacheTimeMap::remove);
   }

}
